package com.atguigu.flink.window;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import com.atguigu.flink.utils.MyUtil;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/4/7
 *
 *     按照类型： 时间，计数
 *    按照特征:  滚动，滑动，会话
 *    按照计算方式: 全局，keyed窗口(非全局)
 *
 *
 *    滚动的，全局，时间窗口
 *    xxxTumblingWindow： 滚动
 *    xxx : keyed window 。每种key都有自己的窗口
 *    windowXxx: 时间窗口
 *
 *   --------------------
 *      flink中有两种时间语义。
 *          ProcessingTime: 取决于计算机的物理时钟。
 *          Event(数据)Time: 取决于某一条数据中携带的时间属性。
 */
public class Demo4_TimeWindow
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        KeyedStream<WaterSensor, String> ds = env
            .socketTextStream("hadoop102", 8888)
            .map(new WaterSensorMapFunction())
            .keyBy(WaterSensor::getId);

        ds
            //5s的滚动窗口，计算机的物理时钟过了5s，就触发一次运算。把过去5s到达的数据放入窗口，进行计算
            //.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
            //slide=3s，size=5s的滑动窗口。 每间隔3s计算一次，计算的窗口的 范围[end-start]是5s
            //.window(SlidingProcessingTimeWindows.of(Time.seconds(5),Time.seconds(3)))
            //每两个数据达到的时间间隔超过5s，上一个窗口就关闭，触发运算，下一个窗口开启
            .window(ProcessingTimeSessionWindows.withGap(Time.seconds(7)))
            .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>()
            {
                @Override
                public void process(String key, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                    TimeWindow window = context.window();

                    out.collect(key +"==>"+MyUtil.parseTimeWindow(window)+":"+MyUtil.parseToList(elements));
                }
            })
            .print();


        try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }
}
